package com.atguigu.gmallrealtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.PropertyNamingStrategy;
import com.alibaba.fastjson.serializer.SerializeConfig;
import com.atguigu.gmallrealtime.app.func.DimMapFunctionHbaseAsync;
import com.atguigu.gmallrealtime.bean.TradeProvinceOrderBean;
import com.atguigu.gmallrealtime.common.Constant;
import com.atguigu.gmallrealtime.util.DateFormatUtil;
import com.atguigu.gmallrealtime.util.DorisUtil;
import com.atguigu.gmallrealtime.util.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.util.concurrent.TimeUnit;

/**
 * @author yhm
 * @create 2023-10-13 14:08
 */
public class DwsTradeProvinceOrderWindow {
    public static void main(String[] args) throws Exception {
        // TODO 1 创建flink环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        // TODO 2 添加检查点和状态后端
        env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE);

        //2.2 设置检查点超时时间
        env.getCheckpointConfig().setCheckpointTimeout(60000L);
        //2.3 设置job取消之后 检查点是否保留
        env.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        //2.4 设置两个检查点之间最小的时间间隔
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000L);
        //2.5 设置重启策略
        // env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,3000L));
        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(30), Time.seconds(3)));

        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");

        System.setProperty("HADOOP_USER_NAME", "atguigu");

        // TODO 3 读取dwd_trade_order_detail主题数据
        String groupId = "dws_trade_province_order_window";
        KafkaSource<String> kafkaSource = MyKafkaUtil.getKafkaSource(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL, groupId);
        DataStreamSource<String> odSource = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "od_source");

        // A left join B
        // A : null
        // null
        // A : B
        // TODO 4 过滤清洗转换数据结构
        // 过滤第二条null
        SingleOutputStreamOperator<JSONObject> jsonObjectStream = odSource.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                try {
                    if (value != null && value.length() != 0) {
                        out.collect(JSONObject.parseObject(value));
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });

        // TODO 5 回撤流数据处理
        // 需要撤回的是第一条数据 A : null
        // 使用替代撤回的方式来实现
        // A : null  和   A - A : B ( 0 : B)
        // 进入窗口累加数据 因为使用的是 第一条 A的信息 会造成 B 中的列为空
        SingleOutputStreamOperator<TradeProvinceOrderBean> beanStream = jsonObjectStream.keyBy(new KeySelector<JSONObject, String>() {
            @Override
            public String getKey(JSONObject value) throws Exception {
                return value.getString("id");
            }
        }).map(new RichMapFunction<JSONObject, TradeProvinceOrderBean>() {
            ValueState<JSONObject> lastOdState;

            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<JSONObject> lastOdDes = new ValueStateDescriptor<>("last_od", JSONObject.class);
                lastOdDes.enableTimeToLive(StateTtlConfig.newBuilder(Time.seconds(20L)).build());
                lastOdState = getRuntimeContext().getState(lastOdDes);
            }

            @Override
            public TradeProvinceOrderBean map(JSONObject value) throws Exception {
                JSONObject lastOd = lastOdState.value();
                TradeProvinceOrderBean.TradeProvinceOrderBeanBuilder builder = TradeProvinceOrderBean.builder();
                if (lastOd != null) {
                    // 当前不是第一条数据
                    builder.orderId(value.getString("order_id"))
                            .provinceId(value.getString("province_id"))
                            .orderAmount(value.getBigDecimal("split_total_amount").subtract(lastOd.getBigDecimal("split_total_amount")))
                            .orderCount(0L)
                            .ts(value.getLong("ts") * 1000);

                } else {
                    // 当前为第一条数据
                    builder.orderId(value.getString("order_id"))
                            .provinceId(value.getString("province_id"))
                            .orderAmount(value.getBigDecimal("split_total_amount"))
                            .orderCount(1L)
                            .ts(value.getLong("ts") * 1000);

                }
                lastOdState.update(value);
                return builder.build();
            }
        });

//        beanStream.print();

       // 注册水位线
        SingleOutputStreamOperator<TradeProvinceOrderBean> withWaterMarkStream = beanStream.assignTimestampsAndWatermarks(WatermarkStrategy.<TradeProvinceOrderBean>forMonotonousTimestamps().withTimestampAssigner(new SerializableTimestampAssigner<TradeProvinceOrderBean>() {
            @Override
            public long extractTimestamp(TradeProvinceOrderBean element, long recordTimestamp) {
                return element.getTs();
            }
        }));
        // TODO 6 按照省份ID进行分组
        // TODO 7 开窗聚合
        WindowedStream<TradeProvinceOrderBean, String, TimeWindow> windowStream = withWaterMarkStream.keyBy(new KeySelector<TradeProvinceOrderBean, String>() {
            @Override
            public String getKey(TradeProvinceOrderBean value) throws Exception {
                return value.getProvinceId();
            }
        }).window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(15L))).allowedLateness(org.apache.flink.streaming.api.windowing.time.Time.seconds(5L));

        SingleOutputStreamOperator<TradeProvinceOrderBean> reduceStream = windowStream.reduce(new ReduceFunction<TradeProvinceOrderBean>() {
            @Override
            public TradeProvinceOrderBean reduce(TradeProvinceOrderBean value1, TradeProvinceOrderBean value2) throws Exception {
                value1.setOrderAmount(value1.getOrderAmount().add(value2.getOrderAmount()));
                value1.setOrderCount(value1.getOrderCount() + value2.getOrderCount());
                // v1里面缺少B的数据  如果需要使用 做非空判断填写即可
                value1.setOrderId(value1.getOrderId() == null ? value2.getOrderId() : value1.getOrderId());

                return value1;
            }
        }, new ProcessWindowFunction<TradeProvinceOrderBean, TradeProvinceOrderBean, String, TimeWindow>() {
            @Override
            public void process(String s, Context context, Iterable<TradeProvinceOrderBean> elements, Collector<TradeProvinceOrderBean> out) throws Exception {
                TimeWindow window = context.window();
                String stt = DateFormatUtil.toYmdHms(window.getStart());
                String edt = DateFormatUtil.toYmdHms(window.getEnd());
                String curDt = DateFormatUtil.toDorisDate(window.getStart());
                for (TradeProvinceOrderBean element : elements) {
                    element.setStt(stt);
                    element.setEdt(edt);
                    element.setCurDate(curDt);
                    out.collect(element);
                }
            }
        });

        // TODO 8 维度关联补充省份表信息
        SingleOutputStreamOperator<TradeProvinceOrderBean> fullBean = AsyncDataStream.unorderedWait(reduceStream, new DimMapFunctionHbaseAsync<TradeProvinceOrderBean>() {
            @Override
            public String getTable() throws Exception {
                return "dim_base_province";
            }

            @Override
            public String getId(TradeProvinceOrderBean value) throws Exception {
                return value.getProvinceId();
            }

            @Override
            public void join(TradeProvinceOrderBean value, JSONObject dim) throws Exception {
                value.setProvinceName(dim.getString("name"));
            }
        }, 60L, TimeUnit.SECONDS);

//        fullBean.print();
        // TODO 9 写出到doris
        fullBean.map(new MapFunction<TradeProvinceOrderBean, String>() {
            @Override
            public String map(TradeProvinceOrderBean value) throws Exception {

                SerializeConfig config = new SerializeConfig();
                config.setPropertyNamingStrategy(PropertyNamingStrategy.SnakeCase);
                return JSON.toJSONString(value, config);
            }
        }).sinkTo(DorisUtil.getDorisSink("dws_trade_province_order_window"));


        // TODO 10 执行任务
        env.execute();

    }
}
